/**
 * Software License, Version 1.0
 * 
 * Copyright 2003 The Trustees of Indiana University.  All rights reserved.
 * 
 *
 *Redistribution and use in source and binary forms, with or without 
 *modification, are permitted provided that the following conditions are met:
 *
 *1) All redistributions of source code must retain the above copyright notice,
 * the list of authors in the original source code, this list of conditions and
 * the disclaimer listed in this license;
 *2) All redistributions in binary form must reproduce the above copyright 
 * notice, this list of conditions and the disclaimer listed in this license in
 * the documentation and/or other materials provided with the distribution;
 *3) Any documentation included with all redistributions must include the 
 * following acknowledgement:
 *
 *"This product includes software developed by the Community Grids Lab. For 
 * further information contact the Community Grids Lab at 
 * http://communitygrids.iu.edu/."
 *
 * Alternatively, this acknowledgement may appear in the software itself, and 
 * wherever such third-party acknowledgments normally appear.
 * 
 *4) The name Indiana University or Community Grids Lab or NaradaBrokering, 
 * shall not be used to endorse or promote products derived from this software 
 * without prior written permission from Indiana University.  For written 
 * permission, please contact the Advanced Research and Technology Institute 
 * ("ARTI") at 351 West 10th Street, Indianapolis, Indiana 46202.
 *5) Products derived from this software may not be called NaradaBrokering, 
 * nor may Indiana University or Community Grids Lab or NaradaBrokering appear
 * in their name, without prior written permission of ARTI.
 * 
 *
 * Indiana University provides no reassurances that the source code provided 
 * does not infringe the patent or any other intellectual property rights of 
 * any other entity.  Indiana University disclaims any liability to any 
 * recipient for claims brought by any other entity based on infringement of 
 * intellectual property rights or otherwise.  
 *
 *LICENSEE UNDERSTANDS THAT SOFTWARE IS PROVIDED "AS IS" FOR WHICH NO 
 *WARRANTIES AS TO CAPABILITIES OR ACCURACY ARE MADE. INDIANA UNIVERSITY GIVES
 *NO WARRANTIES AND MAKES NO REPRESENTATION THAT SOFTWARE IS FREE OF 
 *INFRINGEMENT OF THIRD PARTY PATENT, COPYRIGHT, OR OTHER PROPRIETARY RIGHTS. 
 *INDIANA UNIVERSITY MAKES NO WARRANTIES THAT SOFTWARE IS FREE FROM "BUGS", 
 *"VIRUSES", "TROJAN HORSES", "TRAP DOORS", "WORMS", OR OTHER HARMFUL CODE.  
 *LICENSEE ASSUMES THE ENTIRE RISK AS TO THE PERFORMANCE OF SOFTWARE AND/OR 
 *ASSOCIATED MATERIALS, AND TO THE PERFORMANCE AND VALIDITY OF INFORMATION 
 *GENERATED USING SOFTWARE.
 */
package cgl.narada.service.reliable.impl;

import java.util.Hashtable;
import java.util.Vector;

import cgl.narada.event.EventID;
import cgl.narada.event.NBEvent;
import cgl.narada.event.NBEventTypes;
import cgl.narada.event.TemplateProfileAndSynopsisTypes;
import cgl.narada.event.impl.NBEventGenerator;
import cgl.narada.matching.Profile;
import cgl.narada.service.ServiceException;
import cgl.narada.service.client.ClientService;
import cgl.narada.service.client.EventConsumer;
import cgl.narada.service.client.EventProducer;
import cgl.narada.service.client.NBEventListener;
import cgl.narada.service.client.SessionService;
import cgl.narada.service.qos.QosEvent;
import cgl.narada.service.reliable.events.RdAckInvoiceEntityEvent;
import cgl.narada.service.reliable.events.RdAckResponseServiceInvoiceEvent;
import cgl.narada.service.reliable.events.RdArchivalServiceNotification;
import cgl.narada.service.reliable.events.RdNakInvoiceEntityEvent;
import cgl.narada.service.reliable.events.RdProfileUpdateRequest;
import cgl.narada.service.reliable.events.RdProfileUpdateResponse;
import cgl.narada.service.reliable.events.RdRecoveryEntityRequest;
import cgl.narada.service.reliable.events.RdRecoveryResponse;
import cgl.narada.service.reliable.events.RdRetransmissionServiceEvent;

/** This class is used to keep track of events/archival notifications received
    and also to issue ACKs and to respond to ACKService responses 
    ALL pertaining to a SINGLE entity .

    @author Shrideep Pallickara
    $Date$
    $Revision$
*/

public class SubscriberPercolator implements NBEventListener, RdDebugFlags {

  private int entityId;
  
  private String moduleName = "SubscriberPercolator: ";

  /** eventId --> QosEvent */
  private Hashtable receivedEvents;

  /** templateId --> syncpoint */
  private Hashtable templatesAndSyncs;

  /** templateId --> Vector(ACKS) */
  private Hashtable templatesAndAcks;

  /** templateId --> largest seqNum in recovery response */
  /** This is done to avoid the generation of multiple acks/ack-Responses 
      while a entity is recovering for a given template-id. */
  private Hashtable suppressAcksDuringRecovery;

  private RdEntityActuatorImpl entityActuator;

  private RdCommunicationsMultiplexerImpl rdsMultiplexer;

  private ClientService clientService;

  private EventProducer producer;

  private EventConsumer consumer;

  private InvoicerThread invoicer;
  private String rdsString = "ReliableDeliveryService/";

  
  public SubscriberPercolator(int entityId, 
			      RdEntityActuatorImpl entityActuator)
    throws ServiceException {
    this.entityId = entityId;
    this.entityActuator = entityActuator;
    receivedEvents = new Hashtable();
    templatesAndSyncs = new Hashtable();
    templatesAndAcks = new Hashtable();
    suppressAcksDuringRecovery =  new Hashtable();
    rdsMultiplexer = new RdCommunicationsMultiplexerImpl();
    rdsMultiplexer.registerSubscriberPercolator(this);
    
    initializeSubscriptions();
    invoicer = new InvoicerThread(entityId, producer);
    invoicer.start();
  }

  private void initializeSubscriptions() throws ServiceException {
     /** Subscribe to the right topics and such */
    clientService = SessionService.getClientService(entityId);
    
    if (!clientService.isInitialized()) {
      throw new ServiceException(moduleName + "Client service for entity=" + 
				 entityId +
				 " should have been previously initialized");
    }
    
    String subscription = entityId + "/Invoice";
    Profile profile = 
      clientService.createProfile(TemplateProfileAndSynopsisTypes.STRING, 
				  subscription);
    consumer = clientService.createEventConsumer(this);
    consumer.subscribeTo(profile);

    String subscription2 = entityId + "/ProfileManagement";
    Profile profile2 = 
      clientService.createProfile(TemplateProfileAndSynopsisTypes.STRING, 
				  subscription2);
    consumer.subscribeTo(profile2);
    
    producer = clientService.createEventProducer();

  }

  public void terminateService() throws ServiceException {
    invoicer.terminateService();
  }

  public void issueRecoveryRequest(RdRecoveryEntityRequest recoveryRequest)
    throws ServiceException {
    int templateId = recoveryRequest.getTemplateId();
    
    String subscription = entityId + "/Recovery/Subscriber";
    Profile profile = 
      clientService.createProfile(TemplateProfileAndSynopsisTypes.STRING, 
				  subscription);
    consumer.subscribeTo(profile);
    long recoveryId = recoveryRequest.getRecoveryRequestId();
    
    System.out.println(moduleName + "Will issue recovery request with " 
		       + "recovery identifier = " + recoveryId +"\n\n\n");
    invoicer.manageRecoveryRequest(recoveryRequest);
    invoicer.issueRecoveryRequest(recoveryRequest);
    /*
      NBEvent nbEvent = 
       producer.generateEvent(TemplateProfileAndSynopsisTypes.STRING,
      rdsString + templateId + "/Recovery", 
      recoveryRequest.getBytes());
      
      NBEventGenerator.setEventType(nbEvent, 
      NBEventTypes.RELIABLE_DELIVERY_EXCHANGE);
      producer.publishEvent(nbEvent);
    */
  }


  public void 
  processAckResponseEvent(RdAckResponseServiceInvoiceEvent ackResponse)
    throws ServiceException {
    int templateId = ackResponse.getTemplateId();
    long newSync = ackResponse.getSyncpoint();
    Object templateKey = new Integer(templateId);
    long currentSync = ((Long)templatesAndSyncs.get(templateKey)).longValue();
    
    if (SubscriberPercolator_Debug) {
      System.out.println(moduleName + "Received " + ackResponse);
    }

    if (currentSync > newSync) {
      if (SubscriberPercolator_Debug) {
	System.out.println(moduleName + "CURRENT SYNC = " + currentSync +
			   " NEW SYNC = " + newSync + " ... will reject");
      }
      return;
    }
    
    if (currentSync < newSync) {
      /** Process the sync advance */
      processSyncAdvance(templateId, newSync);
    }

    RdNakInvoiceEntityEvent nakInvoice = null; 
	
    if (ackResponse.containsMissedSequenceInfo()) {
      if (ackResponse.containsSingleSequenceInfo() ) {
	//nakInvoice = new RdNakInvoiceEntityEvent(templateId, 
	//entityId,ackResponse.getMissedSequence() );
	invoicer.processMissedSequence(templateId, 
				       ackResponse.getMissedSequence());
	
      } else {
	//nakInvoice = 
	//  new RdNakInvoiceEntityEvent(templateId, entityId, 
	//			      ackResponse.getMissedSequences() );
	invoicer.processMissedSequences(templateId, 
					ackResponse.getMissedSequences());
	
      }/*end containsSingleSequenceInfo() */
      /** Now that nakInvoice has been initialized, go ahead and issue
	  a retransmission request */
      //issueRetransmissionRequest(nakInvoice);
    } /*end containsMissedSequenceInfo() */

  }

  /** (a)Change the sync value for this template Id
      (b)Proceed to remove any acks that are in the buffer, and have 
      seqNums less than the new sync.
      (c)Proceed to notify the actuator about the sync change. It SHOULD
      be an advance. */
  private void processSyncAdvance(int templateId, long advanceSync) {
    Object templateKey = new Integer(templateId);
    templatesAndSyncs.remove(templateKey);
    templatesAndSyncs.put(templateKey, new Long(advanceSync));

    Vector ackSet = (Vector) templatesAndAcks.get(templateKey);
    /** Proceed to remove any object less than or equal to advanceSync*/
    for (int i=0; i < ackSet.size(); i++) {
      long compareWith = ((Long)ackSet.elementAt(i)).longValue();
      
      if (compareWith <= advanceSync) {
	ackSet.removeElementAt(i);
	i--; /** Since the prev element is gone, the next element will now
		 be at this position. If elementAt(5) goes, 
		 elementPreviouslyAt(6) now at 5 */
      }      
    }
    invoicer.processSyncAdvance(templateId, advanceSync);
    entityActuator.syncPointHasBeenAdvanced(entityId, templateId, advanceSync);
  }

  
  
  public void 
  processArchivalNotification(RdArchivalServiceNotification archNotification) 
    throws ServiceException {
    EventID eventId = archNotification.getEventId();
    int templateId = archNotification.getTemplateId();
    Object templateKey = new Integer(templateId);
    long sequenceNum = archNotification.getSequenceNumber();

    if (SubscriberPercolator_Debug) {
      System.out.println(moduleName + "Received " +  archNotification);
    }

    if (!templatesAndSyncs.containsKey(templateKey)) {
      templatesAndSyncs.put(templateKey, new Long(0));
    }

    if (!templatesAndAcks.containsKey(templateKey)) {
      templatesAndAcks.put(templateKey, new Vector() );
    }
    
    
    /** Check to see if this notification should be discarded. */
    if (notificationToBeDiscarded(archNotification) ) {
      if (receivedEvents.containsKey(eventId)) {
	receivedEvents.remove(eventId);
      }
      return;
    }
    
    /** Chek to see if the arrival of this notification leads to a 
	retransmission request of the corresponding event */
    if (!receivedEvents.containsKey(eventId)) {
      //RdNakInvoiceEntityEvent nakInvoice = 
      //	new RdNakInvoiceEntityEvent(templateId, entityId, sequenceNum);
      //issueRetransmissionRequest(nakInvoice);
      invoicer.processMissedSequence(templateId, sequenceNum);
      return;
    }
    
    /** If it has reached up until this point, release the corresponding
	event and the archival notification, while issuing an acknowledgement
	too */
    Vector ackSet = (Vector) templatesAndAcks.get(templateKey);
    Object seqObject = new Long(sequenceNum);

    if (!ackSet.contains(seqObject) ) {
      ackSet.add(seqObject);
    } else {
      System.out.println(moduleName + seqObject + "Should NOT have been"
			 + " present in the ackSet "); 
    }
    
    QosEvent qosEvent = (QosEvent) receivedEvents.remove(eventId);
    entityActuator.qosEventAvailableForRelease(qosEvent, archNotification);
    issueAcknowledgement(templateId);
  }
  

  /** Check to see if this notification should be discarded. This is either
      because:
      <li>(a) The sequence number is less than the current sync associated
      with the template in questiom.
      <li>(b) The sequence number is contained in the acknowledgement 
      sequence maintained in templatesAndAcks. */
  private boolean 
  notificationToBeDiscarded(RdArchivalServiceNotification archNotification) 
    throws ServiceException {
    EventID eventId = archNotification.getEventId();
    int templateId = archNotification.getTemplateId();
    Object templateKey = new Integer(templateId);
    long sequenceNum = archNotification.getSequenceNumber();

    long templateSyncVal = 
      ((Long)templatesAndSyncs.get(templateKey)).longValue();

    if (templateSyncVal > sequenceNum) {
      /** Since the sync value is greater, the event is obviously a duplicate*/
      return true;
    }
    
    Vector ackSet = (Vector) templatesAndAcks.get(templateKey);
    Object seqObject = new Long(sequenceNum);

    if (ackSet.contains(seqObject) ) {
      return true;
    }

    return false;
  }




  public void 
  processRetransmittedEvent(RdRetransmissionServiceEvent retransmittedEvent) 
    throws ServiceException  {
    entityActuator.processRetransmittedEvent(retransmittedEvent);
    invoicer.processRetransmittedEvent(retransmittedEvent);
  }
  
  
  public void processReceivedEvent(QosEvent qosEvent) {    
    EventID eventId = qosEvent.getEventId();

    if (receivedEvents.containsKey(eventId)) {
      if (SubscriberPercolator_Debug) { 
	System.out.println(moduleName + "Duplicate event " + eventId +
			   "Discarding ");
      }
      return;
    }  
    receivedEvents.put(eventId, qosEvent);
  }



  public void processRecoveryResponse(RdRecoveryResponse recoveryResponse)
    throws ServiceException {
    System.out.println(moduleName + 
		       "Received recovery response with SubscriberInfo");
    invoicer.manageRecoveryResponse(recoveryResponse);
    /** Subscribe the relevant consumer to the set of previously registered 
	profiles */
    entityActuator.processRecoveryResponse(recoveryResponse);
    
    int templateId = recoveryResponse.getTemplateId(); 
    Object templateKey = new Integer(templateId);
    templatesAndSyncs.put(templateKey, new Long(0));
    templatesAndAcks.put(templateKey, new Vector() );

    if (SubscriberPercolator_Debug) {
      System.out.println(moduleName + "Processing ACKResponse contained in" +
			 " the recovery response");
    }

    RdAckResponseServiceInvoiceEvent ackResponse = 
      recoveryResponse.getSubscriberRecoveryInfo().getAckResponse();
    processAckResponseEvent(ackResponse);    
    
  }



  
  
  /** Issues a retransmission request */
  private void issueRetransmissionRequest(RdNakInvoiceEntityEvent nakInvoice)
    throws ServiceException {
    if (SubscriberPercolator_Debug) {
      System.out.println(moduleName + "Issuing Retransmission request " +
			 nakInvoice);
    }
    int templateId = nakInvoice.getTemplateId();
    NBEvent nbEvent = 
      producer.generateEvent(TemplateProfileAndSynopsisTypes.STRING,
			     rdsString + templateId + "/Invoice", 
			     nakInvoice.getBytes());
    
    NBEventGenerator.setEventType(nbEvent, 
				  NBEventTypes.RELIABLE_DELIVERY_EXCHANGE);
    producer.publishEvent(nbEvent);
    
  }

  /** Issue an acknowledgement based on the accumulated acks on a given
      template id */
  private void issueAcknowledgement(int templateId)
    throws ServiceException {
    Object templateKey = new Integer(templateId);
    Vector ackSet = (Vector) templatesAndAcks.get(templateKey);
    
    int numOfAcks = ackSet.size();
    if (numOfAcks == 0) {
      System.out.println(moduleName + "ackSet.size() == 0, not sure if right");
      return;
    }
    
    RdAckInvoiceEntityEvent ackInvoice = 
      invoicer.createAckInvoiceEvent(templateId, ackSet);
    
    if (ackInvoice == null) {
      System.out.println(moduleName + "Not time yet to issue an ACK invoice."+
			 " This was determined in the invoicer thread. Maybe" +
			 " we have just initiated recovery ...");
      return;
    }
    /*long seqNum = 0;
      long[] seqNums = null;
      RdAckInvoiceEntityEvent ackInvoice = null;
      
      
      if (numOfAcks == 1) {
      seqNum = ((Long)ackSet.firstElement()).longValue();
      System.out.println("[" + seqNum + "] \n");
      ackInvoice = new RdAckInvoiceEntityEvent(templateId, entityId, seqNum);
      } else {
      seqNums = new long[numOfAcks];
      for (int i=0; i < numOfAcks; i++) {
      seqNums[i] = ((Long)ackSet.elementAt(i)).longValue();
      }
      ackInvoice = new RdAckInvoiceEntityEvent(templateId, entityId, seqNums);
      
      } //(numOfAcks == 1 )
    */
    if (SubscriberPercolator_Debug) {
      System.out.println(moduleName + "Issuing ACK \n" +  ackInvoice);
    }
    
    NBEvent nbEvent = 
      producer.generateEvent(TemplateProfileAndSynopsisTypes.STRING,
			     rdsString + templateId + "/Invoice", 
			     ackInvoice.getBytes());
    
    NBEventGenerator.setEventType(nbEvent, 
				  NBEventTypes.RELIABLE_DELIVERY_EXCHANGE);
    producer.publishEvent(nbEvent);
    
  }


  /** Used to propagate a profile update request */
  public void 
  issueProfileUpdateRequest(RdProfileUpdateRequest profileUpdateRequest) 
    throws ServiceException {
    int templateId = profileUpdateRequest.getTemplateId();
    NBEvent nbEvent = 
      producer.generateEvent(TemplateProfileAndSynopsisTypes.STRING,
			     rdsString + templateId + "/ProfileManagement", 
			     profileUpdateRequest.getBytes());
    
    NBEventGenerator.setEventType(nbEvent, 
				  NBEventTypes.RELIABLE_DELIVERY_EXCHANGE);
    producer.publishEvent(nbEvent);
  }
  
  
  public void 
  processProfileUpdateResponse(RdProfileUpdateResponse profileUpdateResponse) 
    throws ServiceException {
 
   entityActuator.processProfileUpdateResponse(profileUpdateResponse);
  }


  public void onEvent(NBEvent nbEvent) {
    int eventType = nbEvent.getEventType();
    
    if (eventType == NBEventTypes.NORMAL_EVENT) {
      System.out.println(moduleName + "Received normal event .." + nbEvent +
			 " The SubscriberPercolator should NOT receive this" );
      
    }
    
    if (eventType == NBEventTypes.RELIABLE_DELIVERY_EXCHANGE ) {
      if (SubscriberPercolator_Debug) {
	System.out.println(moduleName + 
			   "Received reliable delivery Xchange.." );
      }
      
      rdsMultiplexer.manageReliableDeliveryExchange(nbEvent);
      return;
    }
    
    System.out.println(moduleName + "Unknown event type [" + eventType + 
		       "] received ");
  }

}








